#ifndef _EPOLL_SOCKET_
#define _EPOLL_SOCKET_
#include "../common/head.h"
#include "../common/omg_type.h"
#include "../common/circular_buffer.h"
#include "msg.h"

using namespace omg;

enum{
	CONN_UNVRIFY = 1,//未验证
	CONN_VRIFY,	//验证OK
	WILL_DEL};	//待删除
#ifndef EPOLLRDHUP
	#define EPOLLRDHUP 0x2000
#endif
struct EPollSocket {
	enum {
		LISTEN_SOCKET, CONNECT_SOCKET, DATA_SOCKET
	};

	EPollSocket(void) {
		fd = 0;
		_lock.init();
		this->_conn_state = CONN_UNVRIFY;
		_send_buffer = new CircularBuffer();
		_send_buffer->init(SOCKET_BUFFER_SIZE,true);
		_recv_buffer = new CircularBuffer();
		_recv_buffer->init(SOCKET_BUFFER_SIZE,true);

	}

	~EPollSocket(void) {
		LOG(INFO)<<"delete epoll_socket";
		delete _send_buffer;
		delete _recv_buffer;
		_send_buffer = NULL;
		_recv_buffer = NULL;
		::epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, fd, NULL);
		close(fd);
	}
	
	void set_index_id(INT index){
		index_id = index;
	}

	INT get_index_id(){
		return index_id;
	}

	void set_world_id(short world_id){
		_world_id = world_id;
	}

	short get_world_id(){
		return this->_world_id;
	}
	INT set_nodelay(bool nodelay) {
		assert(fd > 0);

		INT flag;
		if (nodelay) {
			flag = 1;
		} else {
			flag = 0;
		}

		INT rv = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));

		return rv;
	}

	INT set_reuseaddr(bool reuse) {
		assert(fd > 0);
		INT flag;

		if (reuse) {
			flag = 1;
		} else {
			flag = 0;
		}

		INT rv = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag));

		return rv;
	}

	INT set_blocking(bool blocking) {
		assert(fd > 0);

		int flag = fcntl(fd, F_GETFL);
		if (flag == -1) {
			return -1;
		}

		if (blocking) {
			flag &= (~O_NONBLOCK);
		} else {
			flag |= O_NONBLOCK;
		}

		INT rv = fcntl(fd, F_SETFL, flag);

		return rv;
	}

	int send_msg_im(const MsgBase* base){
		//TODO need more work
		int data_has_send = 0;
		char* data_head = (char*)base;
		//int timeout = 0;
		while(data_has_send < base->msg_size){
			int rst = ::send(fd,data_head+data_has_send,base->msg_size - data_has_send,0);
			if(rst == -1){
				if(errno == EAGAIN){//缓冲区满了。
					continue;
				}
				return -1;
			}
			data_has_send += rst;
		}
		return data_has_send;
	}

	int send_msg(const char* buffer,int buffer_size){
		int data_has_send = 0;
		while(data_has_send < buffer_size){
			int rst = ::send(fd,buffer+data_has_send,buffer_size - data_has_send,0);
			if(rst == -1){
				if(errno == EAGAIN){
					continue;
				}
				return -1;
			}
			data_has_send += rst;
		}
		return data_has_send;
	}

	int send_msg_to_buffer(const char* msg_buffer,int msg_size){
		_lock.lock();
		int rst = _send_buffer->push(msg_buffer,msg_size);
		if(rst == 0){
			epoll_event ev;	
			ev.data.ptr = this;
			ev.events=EPOLLOUT|EPOLLHUP|EPOLLERR|EPOLLRDHUP|EPOLLONESHOT;
	                //修改sockfd上要处理的事件为EPOLLOUT
        	        epoll_ctl(_epoll_fd,EPOLL_CTL_MOD,fd,&ev);
		}
		_lock.unlock();
		return rst;
	}

	void take_msg_from_buffer_to_send(){
		_lock.lock();
		uint32_t data_length = _send_buffer->get_data_length();
		if(data_length <= 0){
			epoll_event event;
			event.data.ptr =this;
			event.events = EPOLLIN|EPOLLHUP|EPOLLERR|EPOLLRDHUP|EPOLLONESHOT;
			epoll_ctl(_epoll_fd,EPOLL_CTL_MOD,fd,&event);		
			_lock.unlock();
			return;
		}
		char* buffer = _send_buffer->peek(data_length);

		if(buffer != NULL){
			uint32_t send_data_length = send_msg((const char*)buffer,data_length);
			if(send_data_length == data_length){//发送完成，取消写事件的注册
				_send_buffer->discard(send_data_length);
				epoll_event event;
				event.data.ptr = this;
				event.events = EPOLLIN|EPOLLHUP|EPOLLERR|EPOLLRDHUP|EPOLLONESHOT;
				epoll_ctl(_epoll_fd,EPOLL_CTL_MOD,fd,&event);		
			}else if(send_data_length < data_length){
				_send_buffer->discard(send_data_length);
				epoll_event event;
				event.data.ptr =this;
				event.events =  EPOLLOUT|EPOLLIN|EPOLLHUP|EPOLLERR|EPOLLRDHUP|EPOLLONESHOT;
				epoll_ctl(_epoll_fd,EPOLL_CTL_MOD,fd,&event);		
			}
			_lock.unlock();
			return;//if rst == -1 task some action			
		}
	}

	int put_recv_msg_to_buff(const char* src_buffer,uint32_t buffer_size){
		_lock.lock();
		if(_recv_buffer->get_data_length() == 0){//if this is a new message
			if(buffer_size >= sizeof(MsgBase)){
				MsgBase* msg_head = (MsgBase*)src_buffer;
				uint32_t msg_size = msg_head->msg_size;
				if(msg_size > SOCKET_BUFFER_SIZE){//error message
					_lock.unlock();
					return -1;
				}
			}
		}

		int rst = _recv_buffer->push(src_buffer,buffer_size);
		_lock.unlock();
		return rst;
	}

	MsgBase* check_receive_msg_complete(){
		_lock.lock();
		if(_recv_buffer->get_data_length() >= sizeof(MsgBase)){
			MsgBase* msg_head = (MsgBase*)_recv_buffer->peek(sizeof(MsgBase));
			if(msg_head == NULL){
				//_recv_buffer->refactor();
				_lock.unlock();
				return NULL;
			}
			uint32_t msg_size = msg_head->msg_size;
			//if total message length is not enough to read ,so going on receive message
			if(msg_size >  _recv_buffer->get_data_length() || msg_size < 0){
				_lock.unlock();
			    return NULL;	
			}
			char* msg_buffer = new char[msg_size];
			int rst = _recv_buffer->pop(msg_buffer,msg_size);
			if(rst == -1){
				LOG(ERROR)<<"Error of read message";
			}
			//_recv_buffer->discard(msg_size);
			_lock.unlock();
			return (MsgBase*)msg_buffer;	
		}
		_lock.unlock();
		return NULL;	
	}




	void set_client_ip_address(sockaddr_in& sin){
		_sin = sin;
		_ip_str = inet_ntoa(_sin.sin_addr);
		_port 	=_sin.sin_port;
	};

	sockaddr_in get_client_ip_address(){
		return _sin;
	}

	std::string get_client_ip_str(){
		return _ip_str;
	}

	void dump_client_ip_info(){
		LOG(INFO)<<"client :"<<_ip_str.c_str()<<":"<<_port;
	}

	void update_socket_state(int state){
		_conn_state = state;
	}

	void update_socket_will_delete(){
		update_socket_state(WILL_DEL);
		epoll_event event;
		event.events =  EPOLLOUT|EPOLLIN|EPOLLHUP|EPOLLERR|EPOLLRDHUP;
        epoll_ctl(_epoll_fd,EPOLL_CTL_MOD,fd,&event);
	}

	int get_socket_state(){
		return _conn_state;
	}

	EPOLL_SOCKET_FD _epoll_fd;
	INT fd;
	INT type;
	INT index_id;
	short _world_id;
	volatile INT	_conn_state;
	std::string 	_ip_str;
	int				_port;
	sockaddr_in		_sin;
	CircularBuffer		*_send_buffer;
	CircularBuffer		*_recv_buffer;
	omg::MutexLock		_lock;
};
#endif

